package com.asap.demo.rule;

import com.asap.demo.model.RuleBean;

import com.asap.demo.model.StandardEvent;
import com.asap.demo.sourcefunc.MysqlSourceFunction;

import com.asap.demo.sourcefunc.MysqlSourceFunction2;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class RuleDemo {

	private static final Logger logger = LoggerFactory.getLogger(RuleDemo.class);


	public static void main(String[] args) {

		try {
//			ParameterTool argsParameter = ParameterTool.fromArgs(args);
//			ParameterTool allParameter = ParameterTool.fromPropertiesFile("/data/asap/flink-1.11.2/custom_conf/application.properties").mergeWith(argsParameter);


			final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
			env.setParallelism(1);
			env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
			env.enableCheckpointing(Time.minutes(3).toMilliseconds(), CheckpointingMode.EXACTLY_ONCE);

			MapStateDescriptor<String, RuleBean> broadcastVar = new MapStateDescriptor<>("rule_broadcast_var", String.class, RuleBean.class);

			BroadcastStream<RuleBean> broadcastStream = env.addSource(new MysqlSourceFunction2())
					.name("mysql source")
					.setParallelism(1)
					.broadcast(broadcastVar);
			Properties browseProperties = new Properties();
			browseProperties.put("bootstrap.servers", "192.168.1.25:9093");
			browseProperties.put("group.id", "temporal");
			browseProperties.put("auto.offset.reset", "latest");
			DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer<String>(
					"flink_pressure_test",
					new SimpleStringSchema(),
					browseProperties
			));
			dataStreamSource.print("====>kafka");
			dataStreamSource
					.map(StandardEvent::parse)
					.filter(new RichFilterFunction<StandardEvent>() {
						@Override
						public boolean filter(StandardEvent event) throws Exception {
							logger.info("====>filter");
							String deviceParentType = event.getField("DEVICE_PARENT_TYPE");
							return deviceParentType.equals("FW") || deviceParentType.equals("WAF");
						}
					})
					.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<StandardEvent>(Time.seconds(10)) {
						@Override
						public long extractTimestamp(StandardEvent event) {
							logger.info("====>extract");
							String createTime = event.getField("CREATE_TIME");
							return Timestamp.valueOf(createTime).getTime();
						}
					})
					.keyBy(new KeySelector<StandardEvent, Object>() {
						@Override
						public Object getKey(StandardEvent event) throws Exception {
							logger.info("====>keyBy");
							return "SRC_PORT";
						}
					})
					.timeWindow(Time.minutes(1))
					.trigger(ContinuousEventTimeTrigger.of(Time.seconds(10)))
					.apply(new RichWindowFunction<StandardEvent, String, Object, TimeWindow>() {
						@Override
						public void apply(Object o, TimeWindow window, Iterable<StandardEvent> input, Collector<String> out) throws Exception {

						}
					}).print();
//			SingleOutputStreamOperator<StandardEvent> standardEventOutputStreamOperator = stringDataStreamSource.map(new RichMapFunction<String, StandardEvent>() {
//				@Override
//				public StandardEvent map(String value) throws Exception {
//					return StandardEvent.parse(value);
//				}
//			})
//					.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<StandardEvent>() {
//
//				Long currentMaxTimeStamp = 0L;
//				// 支持的最大无序时间为10s
//				final Long maxOutOfOrderless = 10000L;
//				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
//
//				@Override
//				public long extractTimestamp(StandardEvent element, long recordTimestamp) {
//					String createTime = element.getField("CREATE_TIME");
//					long timestamp = Timestamp.valueOf(createTime).getTime();
//					currentMaxTimeStamp = Math.max(currentMaxTimeStamp, timestamp);
//					logger.info("key: {}, event_time:{}, currentMaxTimeStamp:[{}|{}], watermark:[{}|{}]",
//							element.getField("CREATE_TIME")+"::"+element.getField("SRC_PORT")+"::"+element.getField("DST_IP"),
//							element.getField("CREATE_TIME"),
//							currentMaxTimeStamp, sdf.format(currentMaxTimeStamp),
//							getCurrentWatermark().getTimestamp(), sdf.format(getCurrentWatermark().getTimestamp()));
//					return timestamp;
//				}
//
//				@Nullable
//				@Override
//				public Watermark getCurrentWatermark() {
//					return new Watermark(currentMaxTimeStamp - maxOutOfOrderless);
//				}
//			});
//
//			SingleOutputStreamOperator<Object> process = standardEventOutputStreamOperator
//					.filter(new RichFilterFunction<StandardEvent>() {
//						@Override
//						public boolean filter(StandardEvent value) throws Exception {
//							logger.info("filter event: {}", value.toString());
//							return value.getField("DEVICE_PARENT_TYPE").equals("FW");
//						}
//					})
//					.keyBy(new KeySelector<StandardEvent, String>() {
//						@Override
//						public String getKey(StandardEvent value) throws Exception {
//							return value.getField("SRC_PORT");
//						}
//					})
//					.window(TumblingEventTimeWindows.of(Time.minutes(15)))
//					.apply(new WindowFunction<StandardEvent, Object, String, TimeWindow>() {
//						@Override
//						public void apply(String s, TimeWindow window, Iterable<StandardEvent> input, Collector<Object> out) throws Exception {
//							HashSet<String> hashSet = new HashSet<>();
//							for (StandardEvent event : input) {
//								hashSet.add(event.getField("DST_IP"));
//							}
//
//							if (hashSet.size() > 1)
//								out.collect(1);
//						}
//					});

//			process.print();



			SingleOutputStreamOperator<Object> processStream = dataStreamSource.connect(broadcastStream)
					.process(new BroadcastProcessFunction<String, RuleBean, Object>() {
						@Override
						public void processElement(String value, ReadOnlyContext ctx, Collector<Object> out) throws Exception {
							ReadOnlyBroadcastState<String, RuleBean> broadcastState = ctx.getBroadcastState(broadcastVar);
							for (Map.Entry<String, RuleBean> entry : broadcastState.immutableEntries()) {
								logger.info("Entry key:{}, entry value:{}", entry.getKey(), entry.getValue());
							}
							out.collect(value);
						}

						@Override
						public void processBroadcastElement(RuleBean value, Context ctx, Collector<Object> out) throws Exception {

						}

						@Override
						public void open(Configuration parameters) throws Exception {
							super.open(parameters);
						}
					});
			processStream.print();
			env.execute("rule demo");
		} catch (IOException e) {
			logger.error("=======properties not found=======");
			e.printStackTrace();
		} catch (Exception e) {
			logger.error("=======execute error=======");
			e.printStackTrace();
		}

	}
}
